package com.sali.service.statistic;

import com.sali.commons.config.ThreadPoolManager;
import com.sali.commons.exception.ProcessChainHandlerInitException;
import com.sali.commons.exception.ProcessChainHandlerProcessException;
import com.sali.entity.ProcessDefinition;
import com.sali.entity.TransferObject;
import com.sali.service.data_processor.chain.IProcessChainHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

// 统计各类 ProcessChainHandler 的性能指标，排查性能卡点
public class PerformanceStatisticProxy implements IProcessChainHandler {

    private static Logger log = LoggerFactory.getLogger(IProcessChainHandler.class);

    // 统计日志输出间隔
    private static final int PRINT_INTERVAL = 10;

    // 统计handler的处理条数
    private static Map<String, AtomicLong> counter = new HashMap<>();

    // 统计handler处理耗时
    private static Map<String, AtomicLong> processCost = new HashMap<>();

    private IProcessChainHandler handler;

    public PerformanceStatisticProxy(IProcessChainHandler handler) {
        this.handler = handler;
        counter.put(this.handler.getClass().getName(), new AtomicLong());
        processCost.put(this.handler.getClass().getName(), new AtomicLong());
    }

    static {
        // 定时执行一次，输出统计指标
        ThreadPoolManager.getScheduledExecutor().scheduleAtFixedRate(PerformanceStatisticProxy::printLog, 0, PRINT_INTERVAL, TimeUnit.SECONDS);
    }

    @Override
    public void preProcess(ProcessDefinition processDefinition) throws ProcessChainHandlerInitException {
        // handler实际逻辑
        this.handler.preProcess(processDefinition);
    }

    @Override
    public void process(TransferObject input) throws ProcessChainHandlerProcessException {
        long start = System.currentTimeMillis();
        // handler实际逻辑
        this.handler.process(input);
        final long cost = System.currentTimeMillis() - start;

        // 异步计数
        CompletableFuture.runAsync(() -> {
            String className = handler.getClass().getName();
            counter.get(className).getAndIncrement();
            processCost.get(className).set(processCost.get(className).get() + cost);
        });
    }

    // 输出统计日志指标
    private static void printLog() {
        for (Map.Entry<String, AtomicLong> entry : counter.entrySet()) {
            String handlerName = entry.getKey();
            AtomicLong num = entry.getValue();
            AtomicLong pCost = processCost.get(handlerName);
            double res = pCost.get() == 0 ? 0 : (num.get() / (pCost.get() * 1.0)) * 1000;
            String logTemplate = "处理器：%s,最近%s秒处理%s条数据，耗时：%s，平均性能%s条/秒";
            if (num.get() != 0) {
                log.info("[性能统计指标] {}", String.format(logTemplate, handlerName, PRINT_INTERVAL, num.get(), pCost.get(), res));
            }
            //重置
            num.set(0);
            pCost.set(0);
        }
    }

}
